package com.junjie.blibli.config;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.junjie.blibli.Constant.RocketMqConstant;
import com.junjie.blibli.Entity.UserFollowing;
import com.junjie.blibli.Entity.UserMoment;
import com.junjie.blibli.service.UserFollowingService;
import com.junjie.blibli.service.webSocket.WebSocketService;
import io.netty.util.internal.StringUtil;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.common.message.MessageExt;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.core.RedisTemplate;

import java.util.ArrayList;
import java.util.List;

@Configuration
public class RocketMqConfig {
    @Value("${rocketmq.name.server.address}")
    private String nameServerAddr;

    @Autowired
    private RedisTemplate<String,String> redisTemplate;

    @Autowired
    private UserFollowingService userFollowingService;

    //跟用户动态相关的生产者
    @Bean("momentsProducer")
    public DefaultMQProducer momentsProducer() throws MQClientException {
        //这里构造函数传的是分组名称
        DefaultMQProducer producer = new DefaultMQProducer(RocketMqConstant.GROUP_MOMENTS);
        producer.setNamesrvAddr(nameServerAddr);
        producer.start();
        return producer;
    }

    //跟用户动态相关的消费者，而且是使用的push方式
    @Bean("momentsConsumer")
    public DefaultMQPushConsumer momentsConsumer() throws MQClientException {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(RocketMqConstant.GROUP_MOMENTS);
        consumer.setNamesrvAddr(nameServerAddr);
        //设置订阅的内容
        //第一个参数是主题，第二个是二级主题，这里写的*就是说主题下所有分类都订阅
        consumer.subscribe(RocketMqConstant.TOPIC_MOMENTS,"*");
        //注册信息监听
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            //MessageExt是在Message基础上拓展了字段，继承自Message，在传出消息基础上又增加了一些处理结果
            //ConsumeConcurrentlyContext是上下文，存放处理过程相关信息
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {

                //因为只发送一条信息，所以直接get(0)
                MessageExt msg = list.get(0);
                //msg为null就什么都不做
                if(msg==null){
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                }
                //这个是避免信息一直重复发的代码，直接骗一波编译器，通过掉
                //因为有些数据直接去数据库里面改，可能会改出问题，导致消息一直没消费掉
//                if(1==1) return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                String bodyStr = new String(msg.getBody());
                //JsonObject就是一个Map，先将字符串转成这个Map，再用toJavaObject并指定类型转换成指定类型
                UserMoment userMoment = JSONObject.toJavaObject(JSONObject.parseObject(bodyStr), UserMoment.class);
                //先获取userId，目的是查找该用户的粉丝
                Long userId = userMoment.getUserId();
                //获取用户粉丝
                List<UserFollowing> fanList = userFollowingService.getUserFans(userId);
                System.out.println();
                //把信息发送给所有用户
                for (UserFollowing fan : fanList) {
                    //这里优化一下，把信息放到redis里面，便于直接查取
                    //哪个粉丝关注了该用户，就作为key值，如下
                    String key = "subscribed-" + fan.getUserId();
                    //该粉丝不只关注了一个人，肯定还关注了很多其他博主，因此这里设置一个列表，redis中用string存
                    String subscribedListStr = redisTemplate.opsForValue().get(key);
                    //将该string转换成一个集合
                    List<UserMoment> subscribedList;
                    //先判断值是否为null，为null就要生成新的空列表，否则就直接转换成指定集合
                    if(StringUtil.isNullOrEmpty(subscribedListStr)){
                        subscribedList = new ArrayList<>();
                    }else{
                        subscribedList = JSONArray.parseArray(subscribedListStr,UserMoment.class);
                    }
                    //不管怎样都要先添加到集合中
                    subscribedList.add(userMoment);
                    //放入redis中
                    subscribedList.add(userMoment);
                    redisTemplate.opsForValue().set(key,JSONObject.toJSONString(subscribedList));
                }
                //返回成功信号
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        consumer.start();
        return consumer;
    }

    //弹幕生产者
    @Bean("danmusProducer")
    public DefaultMQProducer danmusProducer() throws Exception{
        // 实例化消息生产者Producer
        DefaultMQProducer producer = new DefaultMQProducer(RocketMqConstant.GROUP_DANMUS);
        // 设置NameServer的地址
        producer.setNamesrvAddr(nameServerAddr);
        // 启动Producer实例
        producer.start();
        return producer;
    }

    //弹幕消费者
    @Bean("danmusConsumer")
    public DefaultMQPushConsumer danmusConsumer() throws Exception{
        // 实例化消费者
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(RocketMqConstant.GROUP_DANMUS);
        // 设置NameServer的地址
        consumer.setNamesrvAddr(nameServerAddr);
        // 订阅一个或者多个Topic，以及Tag来过滤需要消费的消息
        consumer.subscribe(RocketMqConstant.TOPIC_DANMUS, "*");
        // 注册回调实现类来处理从broker拉取回来的消息
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                MessageExt msg = msgs.get(0);
                byte[] msgByte = msg.getBody();
                String bodyStr = new String(msgByte);
                JSONObject jsonObject = JSONObject.parseObject(bodyStr);
                String sessionId = jsonObject.getString("sessionId");
                String message = jsonObject.getString("message");
                //根据sessionId获取webSocket
                WebSocketService webSocketService = WebSocketService.WEBSOCKET_MAP.get(sessionId);
                if(webSocketService.getSession().isOpen()){
                    try {
                        webSocketService.sendMessage(message);
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }
                // 标记该消息已经被成功消费，这个类是mq直接提供的
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        // 启动消费者实例
        consumer.start();
        return consumer;
    }

}
